package com.hd.assistant.biz.manager;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.hd.assistant.biz.manager.exception.ManagerException;
import com.hd.assistant.biz.manager.ordermining.ItemCostCacheManager;
import com.hd.assistant.biz.manager.ordermining.OrderMiningManager;
import com.hd.assistant.dal.dao.exception.DAOException;
import com.hd.assistant.dal.dao.orderming.ITaskDAO;
import com.hd.assistant.dal.dao.orderming.NotifyMessageDAO;
import com.hd.assistant.dal.dao.orderming.TcOrderDAO;
import com.hd.assistant.dal.dao.query.ordermining.NotifyMessageQuery;
import com.hd.assistant.dal.dataobject.User;
import com.hd.assistant.dal.dataobject.ordermining.ITask;
import com.hd.assistant.dal.dataobject.ordermining.NotifyMessage;
import com.hd.assistant.dal.dataobject.ordermining.TcOrder;
import com.taobao.api.ApiException;
import com.taobao.api.domain.NotifyTopats;
import com.taobao.api.domain.NotifyTrade;
import com.taobao.api.domain.Task;
import com.taobao.api.domain.Trade;
import com.taobao.api.internal.parser.json.ObjectJsonParser;
import com.taobao.api.internal.util.AtsUtils;
import com.taobao.api.response.TradeFullinfoGetResponse;

public class DefaultMessageProcessManager {
	private static String FILE_PATH = "/home/admin/orders/";
	private static String FILE_UNZIP_PATH = "/home/admin/ordersunzip/";
	private static Integer TASK_POOP_SIZE = 4;
	public static Integer TASK_QUENE_SIZE = 7;
	private final static Logger log = LoggerFactory
			.getLogger(MessageMonitor.class);

	/**
	 * 处理tarde消息的线程池
	 */
	private ThreadPoolExecutor tradeThreadPool = new ThreadPoolExecutor(
			TASK_POOP_SIZE, TASK_POOP_SIZE, 0L, TimeUnit.MILLISECONDS,
			new LinkedBlockingQueue<Runnable>(TASK_QUENE_SIZE));
	/**
	 * 处理
	 */
	private ThreadPoolExecutor topAtsThreadPool = new ThreadPoolExecutor(
			TASK_POOP_SIZE, TASK_POOP_SIZE, 0L, TimeUnit.MILLISECONDS,
			new LinkedBlockingQueue<Runnable>(TASK_QUENE_SIZE));;
	@Autowired
	TcOrderDAO tcOrderDAO;
	@Autowired
	NotifyMessageDAO notifyMessageDAO;

	@Autowired
	ITaskDAO iTaskDAO;
	@Autowired
	OrderMiningManager orderMiningManager;
	@Autowired
	ItemCostCacheManager itemCostCacheManager;
	@Autowired
	DefaultEmailSendManager emailSendManager;
	public static final int ITEM_FROM_DB_COUNT = 40; // 每次从数据库取出的数据条数
	public static final int EVERY_TASK_ITEM_COUNT = 10; // 每个任务的记录条数
	public static final long TASK_POOL_FULL_WAIT_TIME = 10000L; // 线程池满了的情况下等待重试的时间
																// (ms)

	private static final int DEFAULT_WAIT_TIME = 2000;// 默认sleep的时间1s
	// private static final double RATE_UP = 1.1; //增长因子
	// private static final double RATE_DOWN = 0.9; //缩小因子
	private static final int DEFAULT_MAX_WAIT_TIME = 5 * 60 * 1000; // 最长等待时间

	private static final String HOST_ID;// 默认采用主机名作为唯一id代表该Host

	private static final Random random = new Random();
	static {
		String hostId = null;
		try {
			hostId = InetAddress.getLocalHost().getHostName();
		} catch (Exception e) {
		}
		HOST_ID = hostId;
	}

	public void init() {
		MessageMonitor monitor = new MessageMonitor();
		new Thread(monitor).start();
		new Thread(new MessageStatusMonitor()).start();
	}

	class MessageStatusMonitor implements Runnable {
		@Override
		public void run() {
			long waitTime = 2 * 3600 * 1000l;
			NotifyMessageQuery query = new NotifyMessageQuery();
			do {
				try {
					Long toId = notifyMessageDAO.getLastProcessedId();

					toId = toId - 500;//修正
					query.setIdTo(toId);
					query.setStatus(NotifyMessage.STATUS_INIT);
					query.setPageSize(ITEM_FROM_DB_COUNT);
					query.setPage(1);
					do {
						query = notifyMessageDAO.getMessageByQuery(query);
						if (query.getMessages() == null
								|| query.getMessages().size() <= 0)
							break;
						// 分发任务逻辑 开始
						List<NotifyMessage> trades = new ArrayList<NotifyMessage>();
						for (NotifyMessage message : query.getMessages()) {
							if (message.getTopic().equals(NotifyMessage.TRADE)) {
								trades.add(message);
							} else if (message.getTopic().equals(
									NotifyMessage.TOPATS)) {
								// ats任务 1个作为处理单元
								AtsMessageProcessedTask task = new AtsMessageProcessedTask(
										message);
								try {
									topAtsThreadPool.execute(task);
								} catch (Exception e) {
									continue;
								}
							}
						}

						generateTradeTask(trades);

						query.setPage(query.nextPage());
					} while (query != null && query.getMessages() != null
							&& query.hasNext());

				} catch (DAOException e1) {
					log.error("DAO EXCEPTION", e1);
				}

				try {
					Thread.sleep(waitTime);
				} catch (InterruptedException e) {
					log.error("InterruptedException ", e);
				}
			} while (true);
		}
	}

	public void generateTradeTask(List<NotifyMessage> trades) {
		int times = trades.size() / EVERY_TASK_ITEM_COUNT;
		if (trades.size() % EVERY_TASK_ITEM_COUNT != 0) {
			times = times + 1;
		}
		for (int i = 0; i < times; i++) {
			int start = i * EVERY_TASK_ITEM_COUNT;
			int to = (i + 1) * EVERY_TASK_ITEM_COUNT > trades.size() ? trades
					.size() : (i + 1) * EVERY_TASK_ITEM_COUNT;
			List<NotifyMessage> t = new ArrayList<NotifyMessage>(
					trades.subList(start, to));
			try {
				tradeThreadPool.execute(new TradeMessageProcessedTask(t));
			} catch (Exception e) {
				log.error("topAtsThreadPool.execute error:", e);
				try {
					Thread.sleep(TASK_POOL_FULL_WAIT_TIME);
				} catch (InterruptedException e1) {
					continue;
				}
				continue;
			}
		}
	}

	class MessageMonitor implements Runnable {

		@Override
		public void run() {
			boolean sleep = false;
			boolean full = false;
			int sleepMiliSecs = DEFAULT_WAIT_TIME;
			List<NotifyMessage> messages = null;
			while (true) {

				if (sleep) {
					log.info("sleep: " + sleepMiliSecs + "miliSec");
					try {
						if (sleepMiliSecs >= DEFAULT_MAX_WAIT_TIME) {
							sleepMiliSecs = DEFAULT_MAX_WAIT_TIME;
						}
						Thread.sleep(sleepMiliSecs);
						sleep = false;
					} catch (Exception e) {
						return;
					}
				}

				// 请求锁
				try {
					if (!notifyMessageDAO.lock(HOST_ID, 600)) {
						// 获取锁失败，随机等待[1-5)s,错开下次请求锁的时间
						random.setSeed(System.currentTimeMillis());
						sleepMiliSecs = random.nextInt(4000) + 1000;
						sleep = true;
						continue;
					}
				} catch (Exception e) {
					log.error("get lock exception.", e);
					// 报异常，等待1秒再试。
					sleepMiliSecs = DEFAULT_WAIT_TIME;
					sleep = true;
					continue;
				}

				try {
					// 取数据逻辑 取完就释放锁
					Long from = notifyMessageDAO.getLastProcessedId();
					List<NotifyMessage> dbMessages = notifyMessageDAO
							.getMessageByLimit(from, ITEM_FROM_DB_COUNT);
					if (dbMessages == null || dbMessages.size() == 0)
					{
						sleep = true;
						sleepMiliSecs = DEFAULT_WAIT_TIME;
						continue;
					}
					Long toIndex = dbMessages.get(dbMessages.size() - 1)
							.getId();
					messages = dbMessages;
					// 更新最新取到哪
					notifyMessageDAO.updateLastProcessedId(toIndex);
				} catch (Exception e) {
					log.error("error:", e);
					sleep = true;
					sleepMiliSecs = DEFAULT_WAIT_TIME;
					continue;
				} finally {
					try {
						if (!notifyMessageDAO.unLock(HOST_ID)) {

						}
					} catch (Exception e2) {
						// ignore me.
					}
				}

				// 分发任务逻辑 开始
				List<NotifyMessage> trades = new ArrayList<NotifyMessage>();
				for (NotifyMessage message : messages) {
					if (message.getTopic().equals(NotifyMessage.TRADE)) {
						trades.add(message);
					} else if (message.getTopic().equals(NotifyMessage.TOPATS)) {
						// ats任务 1个作为处理单元
						AtsMessageProcessedTask task = new AtsMessageProcessedTask(
								message);
						try {
							topAtsThreadPool.execute(task);
						} catch (Exception e) {
							log.error("topAtsThreadPool.execute error:", e);
							try {
								full = true;
								Thread.sleep(TASK_POOL_FULL_WAIT_TIME * 10);
							} catch (InterruptedException e1) {
							}
						}
					}
				}
				if (trades.size() == 0){
					sleep = true;
					sleepMiliSecs = DEFAULT_WAIT_TIME;
					continue;
				}

				generateTradeTask(trades);
				// 分发任务结束

				sleep = true;

				sleepMiliSecs = DEFAULT_WAIT_TIME;
			}

		}
		
		
	}

	
	class TradeMessageProcessedTask implements Runnable {

		List<NotifyMessage> messages;

		public TradeMessageProcessedTask(List<NotifyMessage> messages) {
			super();
			this.messages = messages;
		}

		@Override
		public void run() {
			if (messages == null || messages.size() == 0)
				return;

			for (NotifyMessage message : messages) {
				NotifyTrade notify = (NotifyTrade) message.getMessageObject();
				Trade trade = null;
				TcOrder fromDb;
				try {
					fromDb = orderMiningManager
							.getTcOrderByTid(notify.getTid());
					// 如果是创建和修改价格 则肯定需要请求api 然后存入db
					if (message.getContentStatus().equals(
							NotifyMessage.TradeCreate)) {
						if (fromDb != null) {
							notifyMessageDAO.updateMessageStatus(
									message.getId(),
									NotifyMessage.STATUS_PROCESSED);
							continue;
						}
						trade = orderMiningManager.getOrderAppTradeInfoByTid(
								notify.getTid(), notify.getUserId());
						orderMiningManager.convertTrade(trade, fromDb,
								notify.getUserId());
					}

					if (message.getContentStatus().equals(
							NotifyMessage.TradeModifyFee)
							|| message.getContentStatus().equals(
									NotifyMessage.TradeChanged)) {
						trade = orderMiningManager.getOrderAppTradeInfoByTid(
								notify.getTid(), notify.getUserId());
						if (fromDb != null)
							fromDb.setStatus(null);
						orderMiningManager.convertTrade(trade, fromDb,
								notify.getUserId());
					}
					// 买家已付款 如果本地有 则为了更新paytime 还得重新取
					if (message.getContentStatus().equals(
							NotifyMessage.TradeBuyerPay)) {
						trade = orderMiningManager.getOrderAppTradeInfoByTid(
								notify.getTid(), notify.getUserId());
						orderMiningManager.convertTrade(trade, fromDb,
								notify.getUserId());

					}
					if (message.getContentStatus().equals(
							NotifyMessage.TradeClose)) {
						trade = orderMiningManager.getOrderAppTradeInfoByTid(
								notify.getTid(), notify.getUserId());
						orderMiningManager.convertTrade(trade, fromDb,
								notify.getUserId());

					}

					if (message.getContentStatus().equals(
							NotifyMessage.TradeSuccess)) {
						trade = orderMiningManager.getOrderAppTradeInfoByTid(
								notify.getTid(), notify.getUserId());
						orderMiningManager.convertTrade(trade, fromDb,
								notify.getUserId());
					}

					notifyMessageDAO.updateMessageStatus(message.getId(),
							NotifyMessage.STATUS_PROCESSED);
				} catch (ManagerException e) {
					log.error(" orderMiningManager tid:" + notify.getTid(), e);
					continue;
				} catch (DAOException e) {
					log.error(
							" tcOrderDAO.updateTOrder tid:" + notify.getTid(),
							e);
					continue;
				}
				;

			}
		}

	}

	class AtsMessageProcessedTask implements Runnable {
		NotifyMessage message;

		public AtsMessageProcessedTask(NotifyMessage message) {
			super();
			this.message = message;
		}

		@Override
		public void run() {
			NotifyTopats topats = (NotifyTopats) message.getMessageObject();
			if (!message.getContentStatus().equals("TaskComplete"))
				return;
			try {
				ITask fromDbTask = iTaskDAO.getTaskByTaskId(topats.getTaskId());
				if (fromDbTask == null){
					notifyMessageDAO.updateMessageStatus(message.getId(),
							NotifyMessage.STATUS_PROCESSED);
					return;
				}
				Task task = orderMiningManager.getTaskInfoByTaskId(topats
						.getTaskId());
				if (task == null)
				{
					notifyMessageDAO.updateMessageStatus(message.getId(),
							NotifyMessage.STATUS_PROCESSED);
					return;
				}
				String url = task.getDownloadUrl();
				String path = FILE_PATH + task.getTaskId() + "/";
				File taskFile = AtsUtils.download(url, new File(path)); // 下载文件到本地
				File resultFile = new File(FILE_UNZIP_PATH + task.getTaskId()); // 解压后的结果文件夹
				if (!(resultFile.exists()) && !(resultFile.isDirectory())) {
					resultFile.mkdirs();
				}
				List<File> files = AtsUtils.unzip(taskFile, resultFile); // 解压缩并写入到指定的文件夹
				ObjectJsonParser<TradeFullinfoGetResponse> parse = new ObjectJsonParser(
						TradeFullinfoGetResponse.class);
				for (File f : files) {
					BufferedReader bf = new BufferedReader(new InputStreamReader(new FileInputStream(f), "UTF-8"));
					String s = "";
					while ((s = bf.readLine()) != null) {
						try {
							TradeFullinfoGetResponse res = parse.parse(s);
							if (res == null)
								continue;
							Trade t = res.getTrade();
							TcOrder fromDb = orderMiningManager
									.getTcOrderByTid(t.getTid());
							if (fromDb != null)
								continue;
							orderMiningManager.convertTrade(t, null,
									fromDbTask.getUserId());
						} catch (ApiException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
					}
					bf.close();
				}

				ITask update = new ITask();
				update.setTaskId(task.getTaskId());
				update.setDownloadUrl(task.getDownloadUrl());
				update.setStatus(task.getStatus());
				iTaskDAO.updateTask(update);

				notifyMessageDAO.updateMessageStatus(message.getId(),
						NotifyMessage.STATUS_PROCESSED);
				User u = itemCostCacheManager.getOrderMiningUser(fromDbTask.getUserId());
				if(u != null && u.getEmail() != null)
					emailSendManager.sendDumpTradesOkEmail(u.getEmail());
			} catch (ManagerException e) {
				log.error(" AtsMessageProcessedTask ManagerException taskid= "
						+ topats.getTaskId(), e);
			} catch (DAOException e) {
				log.error(" AtsMessageProcessedTask DAOException taskid= "
						+ topats.getTaskId(), e);
			} catch (ApiException e) {
				log.error(" AtsMessageProcessedTask ApiException taskid= "
						+ topats.getTaskId(), e);
			} catch (IOException e) {
				log.error(" AtsMessageProcessedTask IOException taskid= "
						+ topats.getTaskId(), e);
			}
		}

	}
}
